#include <process.h>

#include <windows.h>
#include <winsock.h>

#include "lib.h"
#include "patch.h"
#include "portio.h"
#include "log.h"

struct PortUserOverlapped
{
	OVERLAPPED	ov;
	PortUser*	user;

	PortUserOverlapped(PortUser* pu)
		{ ZeroMemory(&ov, sizeof(OVERLAPPED)); user = pu; }
	~PortUserOverlapped()
		{ ASSERT(1 == HasOverlappedIoCompleted(&ov)); }
};

PortUser::PortUser()
{
	// handle MUST be set by subclass
	//
	handle = INVALID_HANDLE_VALUE;
	over = (LPOVERLAPPED)new PortUserOverlapped(this);
}

PortUser::~PortUser()
{
	// handle MUST be closed by subclass
	//
	handle = INVALID_HANDLE_VALUE;
	delete (PortUserOverlapped*)over;
	over = (LPOVERLAPPED)0;
}

bool8
PortUser::initOk()
{
	return (bool8)(over != (LPOVERLAPPED)0);	// handle checked by subclass
}

PortIO::PortIO(int32 nWorkers)
{
	port = (HANDLE)0;
	numWorkers = 0;

	workers = new HANDLE[nWorkers];
	if (!workers)
		return;
	ZeroMemory(workers, nWorkers * sizeof(HANDLE));

	port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, KEY_SHUTDOWN, 0);
	if (!port)
		return;

	numWorkers = nWorkers;

	if (!start())
		Log::msgError("PortIO::PortIO: start failed  ");
}

PortIO::~PortIO()
{
	stop();

	delete [] workers;
	workers = (HANDLE*)0;

	if (port)
	{
		CloseHandle(port);
		port = (HANDLE)0;
	}
}

bool8
PortIO::initOk()
{
	return (bool8)(workers && port && numWorkers);
}

bool8
PortIO::start()
{
	DWORD threadID;
	int32 i, abort;

	for (i = 0, abort = -1; abort == -1 && i < numWorkers; i++)
	{
		workers[i] = (HANDLE)_beginthreadex(
			(void*)0, 0, (unsigned(__stdcall*)(void*))workerProc,
			(void*)this, 0, (unsigned*)&threadID);
		if (workers[i] == (HANDLE)0)
			abort = i;
	}

	if (abort != -1)
	{
		if (abort > 0)
		{
			BOOL postedOk = TRUE;

			for (i = 0; postedOk && i < abort; i++)
				postedOk = PostQueuedCompletionStatus(port, 0,
					KEY_SHUTDOWN, (LPOVERLAPPED)0);

			if (postedOk)
			{
				if (WAIT_TIMEOUT ==
					WaitForMultipleObjects(abort, workers, TRUE, 10000))
				{
					Log::msgError("PortIO::start:  WaitFor(workers) failed  ");
					for (i = 0; i < abort; i++)
					{
						TerminateThread(workers[i], 0);
						workers[i] = (HANDLE)0;
					}
				}
			}
			else
			{
				Log::msgError("PortIO::start:  PostQueuedCompletionStatus "
					"failed  (%ld)  ", GetLastError());
				for (i = 0; i < abort; i++)
				{
					TerminateThread(workers[i], 0);
					workers[i] = (HANDLE)0;
				}
			}

			for (i = 0; i < abort; i++)
			{
				CloseHandle(workers[i]);
				workers[i] = (HANDLE)0;
			}
		}
		numWorkers = 0;

		return false;
	}

	return true;
}

void
PortIO::stop()
{
	if (workers && port && numWorkers > 0)
	{
		int32 i;
		BOOL postedOk = TRUE;

		for (i = 0; postedOk && i < numWorkers; i++)
			postedOk = PostQueuedCompletionStatus(port, 0, KEY_SHUTDOWN,
				(LPOVERLAPPED)0);

		if (postedOk)
		{
			if (WAIT_TIMEOUT ==
				WaitForMultipleObjects(numWorkers, workers, TRUE, 10000))
			{
				Log::msgError("PortIO::stop:  WaitFor(workers) failed  ");
				for (i = 0; i < numWorkers; i++)
				{
					TerminateThread(workers[i], 0);
					workers[i] = (HANDLE)0;
				}
			}
		}
		else
		{
			Log::msgError("PortIO::stop:  PostQueuedCompletionStatus "
				"failed  (%ld)  ", GetLastError());
			for (i = 0; i < numWorkers; i++)
			{
				TerminateThread(workers[i], 0);
				workers[i] = (HANDLE)0;
			}
		}

		for (i = 0; i < numWorkers; i++)
		{
			CloseHandle(workers[i]);
			workers[i] = (HANDLE)0;
		}
		numWorkers = 0;
	}
}

bool8
PortIO::newPortUser(PortUser* User)
{
	HANDLE result = port;
	
	if (result)
		result = CreateIoCompletionPort(User->handle, result, KEY_USERIO, 0);

	return (bool8)(result != (HANDLE)0);
}

uint32 __stdcall
PortIO::workerProc(PortIO* This)
{
	bool8			workerRunning = true;
	DWORD			xfer, key;
	LPOVERLAPPED	over;
	int32			result, err;

	while (workerRunning)
	{
		result = GetQueuedCompletionStatus(This->port,
			&xfer, &key, &over, INFINITE);
		err = GetLastError();

		if (result != 0)		// successful
		{
			if (key == KEY_SHUTDOWN)
				workerRunning = false;
			else if (key == KEY_USERIO)
				((PortUserOverlapped*)over)->user->ioDone(xfer);
			else if (key == KEY_TIMER)
				((PortUserOverlapped*)over)->user->ioTimer(xfer);
			else if (key == KEY_CLOSED)
				((PortUserOverlapped*)over)->user->ioDone(xfer);
			else
				Log::msgWarning("worker: GQCS bad key (%ld)", key);
		}
		else
		{
			if (over)			// failed IO (IO packet dequeued)
			{
				if (key == KEY_SHUTDOWN)
					workerRunning = false;
				else if (key == KEY_USERIO)
					((PortUserOverlapped*)over)->user->ioErr(err);
				else if (key == KEY_TIMER)
					;	//ignore timer
				else if (key == KEY_CLOSED)
					((PortUserOverlapped*)over)->user->ioDone(xfer);
				else
					Log::msgWarning("worker: GQCS io error (%ld) bad key (%ld)",
						err, key);
			}
			else				// other error
				Log::msgWarning("worker:  GQCS error (%ld)  ", err);
		}
	}

	_endthreadex(0);
	return 0;
}

void
PortUser::getClientIP(char* buf, int32 maxBufSizeIncludingNull)
{
	SOCKADDR_IN addr;
	int len = sizeof(SOCKADDR_IN);
	char* ip = (char*)0;

	if (0 == getpeername((SOCKET)handle, (SOCKADDR*)&addr, &len))
		ip = inet_ntoa(addr.sin_addr);

	if (ip)
	{
		char* p = ip;

		while (*p++)
			/* find end */ ;

		len = (p - ip - 1);
		if (len >= maxBufSizeIncludingNull)
			len = maxBufSizeIncludingNull - 1;
		CopyMemory(buf, ip, len);
		buf[len] = '\0';
	}
	else
		*buf = '\0';
}
